-module(emqx_plugin_kafka_app).
-behaviour(application).
-emqx_plugin(?MODULE).
-export([
    start/2,
    stop/1
]).

start(_StartType, _StartArgs) ->
    {ok, Sup} = emqx_plugin_kafka_sup:start_link(),
    Cnf = get_kafka_config(),
    emqx_plugin_kafka:load(Cnf),
    {ok, Sup}.

stop(_State) ->
    emqx_plugin_kafka:unload().
    

%  配置文件中获取kafka 配置
get_kafka_config() ->
    AbsolutePath = "/etc/emqx_plugin_kafka.conf",
    Cnf = 
        case filelib:is_file(AbsolutePath) of 
            true -> 
                {ok, Parsed} = hocon:load(AbsolutePath),
                logger:info("get kafka config content: ~p !", [Parsed]),
                % #{<<"plugin_kafka">> => #{<<"bootstrap_hosts">> => [<<"10.1.86.200:30092">>],
                % <<"mqtt_topics">> => [<<"test/#">>],<<"query_api_versions">> => true,
                % <<"reconnect_cool_down_seconds">> => 15,<<"topic">> => <<"test">>}} ! mfa: undefined
                case maps:get(<<"plugin_kafka">>, Parsed, undefined) of
                    undefined ->    
                        logger:error("get kafka config error: ~p !", [AbsolutePath]),
                        fallback_config();
                    KafkaConfig -> 
                        % TODO 这里获取kafka 配置 设置到环境 key kafka , value  是 Config
                        case maps:get(<<"bootstrap_hosts">>, KafkaConfig, undefined) of 
                            undefined -> 
                                logger:error("missing bootstrap_hosts in kafka_config"),
                                fallback_config();
                            _ -> 
                                Config = #{
                                    bootstrap_hosts => maps:get(<<"bootstrap_hosts">>, KafkaConfig, []), 
                                    reconnect_cool_down_seconds  => maps:get(<<"reconnect_cool_down_seconds">>, KafkaConfig, undefined), 
                                    query_api_versions => maps:get(<<"query_api_versions">>, KafkaConfig, undefined),
                                    topic => maps:get(<<"topic">>, KafkaConfig, undefined),
                                    mqtt_topics => maps:get(<<"mqtt_topics">>, KafkaConfig, undefined)
                                },
                                logger:info("env Config: ~p !", [Config]), 
                                application:set_env(emqx_plugin_kafka, kafka_config, Config),
                                Config
                        end       
                end;        

            false ->
                logger:error("get kafka config error: ~p !", [AbsolutePath]),
                fallback_config()
        end,
    Cnf.


% 第二种环境变量获取方式
fallback_config() ->
    BootstrapHosts = case os:getenv("KAFKA_BOOTSTRAP_HOSTS") of
        false -> 
            logger:error("KAFKA_BOOTSTRAP_HOSTS not found in env !"),
            [];
        "" -> 
            logger:error("KAFAKA_BOOTSTRAP_HOSTS is empty !"),
            [];
        Value -> Value    
    end,
    logger:info("env KAFKA_BOOTSTRAP_HOSTS: ~p !", [BootstrapHosts]), 
    Config = #{
        bootstrap_hosts => list_to_binary(BootstrapHosts),
        reconnect_cool_down_seconds => list_to_binary(
            case os:getenv("KAFKA_RECONNECT_COOL_DOWN_SECONDS") of 
                false -> "10";
                Val -> Val
            end 
        ),
        query_api_versions => list_to_binary(
            case os:getenv("KAFKA_QUERY_API_VERSIONS") of 
                false -> "true";
                Val -> Val          
            end    
        ),
        topic => list_to_binary(
            case os:getenv("KAFKA_TOPIC") of 
                false -> "emqx_topic";
                Val -> Val
                
            end    
        ),
        mqtt_topics => case os:getenv("KAFKA_MQTT_TOPICS") of 
            false -> undefined;
            Val -> list_to_binary(Val)
        end
    },
    application:set_env(emqx_plugin_kafka, kafka_config, Config),
    Config.

